package com.region.http.client.pool;

import com.region.common.adapter.logging.LoggerAdapter;
import com.region.common.adapter.logging.LoggerAdapterFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 26225
 * @date 2024/9/1 11:56
 * @description
 */
public class RegionPoolExecutor {

    private final ExecutorService poolExecutor;

    private final int CORE_NUMBER = Runtime.getRuntime().availableProcessors();

    private final int MAX_NUMBER = CORE_NUMBER * 2;

    private final long KEEPALIVE_TIME = 10;

    private final int QUEUE_SIZE = CORE_NUMBER;

    private final LoggerAdapter loggerAdapter = LoggerAdapterFactory.getLogger(getClass());

    public RegionPoolExecutor(ExecutorService poolExecutor) {
        this.poolExecutor = poolExecutor == null ? defaultPoolExecutor() : poolExecutor;
    }

    /**
     * The region client asynchronous pool executor
     * @return
     */
    private ThreadPoolExecutor defaultPoolExecutor() {
        BlockingQueue queue = null;
        if (QUEUE_SIZE > 0) {
            queue = new ArrayBlockingQueue(QUEUE_SIZE);
        } else {
            queue = new SynchronousQueue();
        }
        return new ThreadPoolExecutor(CORE_NUMBER, MAX_NUMBER, KEEPALIVE_TIME, TimeUnit.SECONDS, queue,
                    new ThreadFactory() {
                        AtomicInteger threadNumber = new AtomicInteger(1);
                        ThreadGroup group = (System.getSecurityManager() != null) ? System.getSecurityManager().getThreadGroup() :
                                Thread.currentThread().getThreadGroup();

                        @Override
                        public Thread newThread(Runnable r) {
                            Thread t = new Thread(group, r,
                                    "Region-Client-Asynchronous-Pool-" + threadNumber.getAndIncrement(),
                                    0);
                            if (t.isDaemon())
                                t.setDaemon(false);
                            if (t.getPriority() != Thread.NORM_PRIORITY)
                                t.setPriority(Thread.NORM_PRIORITY);
                            return t;
                        }
                    }, new ThreadPoolExecutor.DiscardPolicy());
    }

    public ExecutorService getPoolExecutor() {
        return poolExecutor;
    }

    public void execute(AsyncRunnable asyncRunnable) {
        poolExecutor.execute(() -> {
            try {
                asyncRunnable.apply();
            } catch (Throwable e) {
                loggerAdapter.error("The asynchronous task execute fail.", e);
            }
        });
    }

    public Future submit(AsyncCallable asyncCallable) {
        return poolExecutor.submit(() -> {
            try {
                return asyncCallable.apply();
            } catch (Throwable e) {
                loggerAdapter.error("The asynchronous task execute fail.", e);
            }
            return null;
        });
    }

    /**
     * Execute the task using CompletableFuture and return the object against it
     *
     * @param asyncCallable
     * @return
     */
    public CompletableFuture submitByCompletable(AsyncCallable asyncCallable) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return asyncCallable.apply();
            } catch (Throwable e) {
                loggerAdapter.error("The execute completable future fail.", e);
            }
            return null;
        }, this.poolExecutor);
    }

    /**
     * Closed region pool executor
     */
    public void shutdown() {
        if (this.poolExecutor != null && !this.poolExecutor.isShutdown()) {
            this.poolExecutor.shutdown();
        }
    }
}
